package com.atguigu.gmall.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.bean.DimSkuInfoBean;
import com.atguigu.gmall.bean.TradeSkuOrderBean;
import com.atguigu.gmall.util.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

/**
 * @author yhm
 * @create 2022-08-27 9:22
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取kafka对应主题的数据dwd_trade_order_detail
        String topicName = "dwd_trade_order_detail";
        String groupID = "dws_trade_sku_order_window";
        DataStreamSource<String> orderDetailStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));

        // TODO 4 转换结构
        // TODO 5 过滤掉不完整的数据
        SingleOutputStreamOperator<JSONObject> jsonObjStream = orderDetailStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String userId = jsonObject.getString("user_id");
                String sourceType = jsonObject.getString("source_type");
                if (userId != null && sourceType != null) {
                    out.collect(jsonObject);
                }
            }
        });

        // TODO 6 按照订单详情分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("id");
            }
        });

        // TODO 7 去重因为left join造成的重复数据  (撤回流发送的null数据已经被上一级过滤掉)
        SingleOutputStreamOperator<JSONObject> processStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

            ValueState<JSONObject> lastOrderJsonObjState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastOrderJsonObjState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("last_order_jsonObj", JSONObject.class));
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject lastOrderJsonObj = lastOrderJsonObjState.value();
                if (lastOrderJsonObj == null) {
                    // 当前id的第一条数据
                    lastOrderJsonObjState.update(value);
                    long currentProcessingTime = ctx.timerService().currentProcessingTime();
                    ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                } else {
                    // 后续的数据
                    // 比较join的时间  2022-08-26 07:47:30.513Z
                    String rowOpTs = value.getString("row_op_ts");
                    String lastRowOpTs = lastOrderJsonObj.getString("row_op_ts");
                    if (TimestampLtz3CompareUtil.compare(rowOpTs,lastRowOpTs) >= 0) {
                        lastOrderJsonObjState.update(value);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                // 输出状态中的值
                JSONObject lastOrderJsonObj = lastOrderJsonObjState.value();
                out.collect(lastOrderJsonObj);
                lastOrderJsonObjState.clear();
            }
        });

        // TODO 8 根据sku来聚合之后再开窗累加
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = processStream.map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(JSONObject value) throws Exception {
                return TradeSkuOrderBean.builder()
                        .skuId(value.getString("sku_id"))
                        .skuName(value.getString("sku_name"))
                        .originalAmount(value.getDouble("split_original_amount") == null ? 0.0: value.getDouble("split_original_amount"))
                        .activityAmount(value.getDouble("split_activity_amount") == null ? 0.0 : value.getDouble("split_activity_amount"))
                        .couponAmount(value.getDouble("split_coupon_amount") == null ? 0.0 : value.getDouble("split_coupon_amount"))
                        .orderAmount(value.getDouble("split_total_amount") == null ? 0.0 : value.getDouble("split_total_amount"))
                        .ts(value.getLong("od_ts") * 1000L)
                        .build();
            }
        });

        SingleOutputStreamOperator<TradeSkuOrderBean> beanWithWatermarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeSkuOrderBean>() {
            @Override
            public long extractTimestamp(TradeSkuOrderBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        SingleOutputStreamOperator<TradeSkuOrderBean> reduceBeanStream = beanWithWatermarkStream.keyBy(TradeSkuOrderBean::getSkuId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                        value1.setActivityAmount(value1.getActivityAmount() + value2.getActivityAmount());
                        value1.setCouponAmount(value1.getCouponAmount() + value2.getCouponAmount());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                }, new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean tradeSkuOrderBean = input.iterator().next();
                        tradeSkuOrderBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        tradeSkuOrderBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        tradeSkuOrderBean.setTs(System.currentTimeMillis());
                        out.collect(tradeSkuOrderBean);
                    }
                });


        // TODO 9 和DIM层的维度表进行关联
        // 和 DIM_SKU_INFO 关联
        reduceBeanStream.map(new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
            DruidDataSource dataSource = null;
            @Override
            public void open(Configuration parameters) throws Exception {
                dataSource = DruidPhoenixDSUtil.getDataSource();
            }

            @Override
            public TradeSkuOrderBean map(TradeSkuOrderBean value) throws Exception {
                DruidPooledConnection connection = dataSource.getConnection();
                // select * from t where id = 'v.id'


                // 关联sku商品表
                JSONObject skuInfo = DimUtil.getDimInfo(connection, "DIM_SKU_INFO", value.getSkuId());

                value.setSkuName(skuInfo.getString("skuName"));
                value.setSpuId(skuInfo.getString("spuId"));
                value.setCategory3Id(skuInfo.getString("category3Id"));


                // 关联spu商品表
                // spu的id

                return value;
            }
        }).print();

        // TODO 10 写出数据到clickHouse中

        // TODO 11 执行任务
        env.execute();

    }
}
